package drds.data_propagate.parse;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.LifecycleAware;
import drds.data_propagate.binlog.BinLogContext;
import drds.data_propagate.binlog.BinLogEvent;
import drds.data_propagate.binlog.Buffer;
import drds.data_propagate.binlog.Decoder;
import drds.data_propagate.binlog.event.rows_event.DeleteRowsEvent;
import drds.data_propagate.binlog.event.rows_event.UpdateRowsEvent;
import drds.data_propagate.binlog.event.rows_event.WriteRowsEvent;
import drds.data_propagate.parse.exception.ParseException;
import drds.data_propagate.parse.table_meta_data.TableMetaData;
import drds.propagate.protocol.Entry;

class ParserStage implements EventHandler<MessageEvent>, LifecycleAware {

    private MultiStageCoprocessorImpl multiStageCoprocessor;
    private Decoder decoder;
    private BinLogContext context;

    public ParserStage(MultiStageCoprocessorImpl multiStageCoprocessor, BinLogContext context) {
        this.multiStageCoprocessor = multiStageCoprocessor;
        decoder = new Decoder(BinLogEvent.UNKNOWN_EVENT, BinLogEvent.ENUM_END_EVENT);
        this.context = context;
        if (multiStageCoprocessor.gtidSet != null) {
            context.setGtidSet(multiStageCoprocessor.gtidSet);
        }
    }

    public void onEvent(MessageEvent messageEvent, long index, boolean endOfBatch) throws Exception {
        try {
            BinLogEvent binLogEvent = messageEvent.getBinLogEvent();
            if (binLogEvent == null) {
                Buffer buffer = messageEvent.getBuffer();
                binLogEvent = decoder.decode(buffer, context);
                messageEvent.setBinLogEvent(binLogEvent);
            }

            int eventType = binLogEvent.getHeader().getEventType();
            TableMetaData tableMetaData = null;
            boolean needDmlParse = false;
            switch (eventType) {
                case BinLogEvent.WRITE_ROWS_EVENT_V1:
                case BinLogEvent.WRITE_ROWS_EVENT:
                    tableMetaData = multiStageCoprocessor.logEventConvert.parseRowsEventForTableMeta((WriteRowsEvent) binLogEvent);
                    needDmlParse = true;
                    break;
                case BinLogEvent.UPDATE_ROWS_EVENT_V1:
                case BinLogEvent.PARTIAL_UPDATE_ROWS_EVENT:
                case BinLogEvent.UPDATE_ROWS_EVENT:
                    tableMetaData = multiStageCoprocessor.logEventConvert.parseRowsEventForTableMeta((UpdateRowsEvent) binLogEvent);
                    needDmlParse = true;
                    break;
                case BinLogEvent.DELETE_ROWS_EVENT_V1:
                case BinLogEvent.DELETE_ROWS_EVENT:
                    tableMetaData = multiStageCoprocessor.logEventConvert.parseRowsEventForTableMeta((DeleteRowsEvent) binLogEvent);
                    needDmlParse = true;
                    break;
                case BinLogEvent.ROWS_QUERY_LOG_EVENT:
                    needDmlParse = true;
                    break;
                default:
                    Entry entry = multiStageCoprocessor.logEventConvert.parse(messageEvent.getBinLogEvent(), false);
                    messageEvent.setEntry(entry);
            }

            // 记录一下DML的表结构
            messageEvent.setNeedDmlParse(needDmlParse);
            messageEvent.setTableMetaData(tableMetaData);
        } catch (Throwable e) {
            multiStageCoprocessor.exception = new ParseException(e);
            throw multiStageCoprocessor.exception;
        }
    }

    @Override
    public void onStart() {

    }

    @Override
    public void onShutdown() {

    }
}
